package com.example.combine;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * Created with IntelliJ IDEA.
 * ClassName: UnionDemp
 * Package: com.example.combine
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-19
 * Time: 9:55
 */

public class ConnectDemo {
    public static void main(String[] args) throws Exception {
        //总结
        //使用connect 一次只能连接两条流
        //流的数据类型可以不一样
        //连接之后 map flatmap process各自处理各自的流
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);
        //创建两条流
//        DataStreamSource<Integer> source1 = env.fromElements(1,2,3);
//        DataStreamSource<String> source2 = env.fromElements("aa", "bb", "cc");
        //创建socket
        SingleOutputStreamOperator<Integer> source1 =
                env.socketTextStream("hadoop102", 7777).map(val -> Integer.parseInt(val));
        //如果在这里输出字符 那么直接 For input string: "abc" 程序退出
        // socketTestStream 默认的返回类型是DataStreamSource<String>
        DataStreamSource<String> source2 =
                env.socketTextStream("hadoop102", 9999);
        //连接流 没有做数据转换 也直接传值了 而且一次只能传一个
        ConnectedStreams<Integer, String> connect = source1.connect(source2);
        //返回值也是一个连接流
        //CoMapFunction三个参数 分别是 第一条流的类型 第二条流的类型 输出的类型
        SingleOutputStreamOperator<String> map = connect.map(new CoMapFunction<Integer, String, String>() {
            //谁调用connect 谁就是1
            @Override
            public String map1(Integer value) throws Exception {
                return "数字流" + value.toString();
            }

            @Override
            public String map2(String value) throws Exception {
                return  "字符流" + value;
            }
        });//返回的是流数据
        map.print();
        env.execute();
    }
}
